package com.cjkj.equipment.utils;

import com.cjkj.equipment.config.AMQPConfig;
import javafx.application.Application;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.codec.binary.Base64;
import org.apache.qpid.jms.JmsConnection;
import org.apache.qpid.jms.JmsConnectionListener;
import org.apache.qpid.jms.message.JmsInboundMessageDispatch;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;
import javax.jms.*;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;
import java.net.URI;
import java.util.Hashtable;

/**
 * AMQP工具类
 * @author cjkj
 * @date 2024-08-30
 */
@Slf4j
public class AMQPUtils {
    final static Logger logger = LoggerFactory.getLogger(Application.class);
    public static void init(String consumerGroupId, AMQPConfig amqp, MessageListener listener) throws Exception {
        for (int i = 0; i < amqp.getConnectionCount(); i++) {
            long timeStamp = System.currentTimeMillis();
            String signMethod = "hmacsha1";
            String userName = amqp.getClientId() + "-" + i + "|authMode=aksign"
                    + ",signMethod=" + signMethod
                    + ",timestamp=" + timeStamp
                    + ",authId=" + amqp.getAccessKey()
                    + ",iotInstanceId=" + amqp.getIotInstanceId()
                    + ",consumerGroupId=" + consumerGroupId
                    + "|";
            String signContent = "authId=" + amqp.getAccessKey() + "&timestamp=" + timeStamp;
            String password = doSign(signContent, amqp.getAccessSecret(), signMethod);
            Context context = getContext(amqp);
            ConnectionFactory cf = (ConnectionFactory)context.lookup("SBCF");
            Destination queue = (Destination)context.lookup("QUEUE");
            // 创建连接。
            Connection connection = cf.createConnection(userName, password);
            ((JmsConnection)connection).addConnectionListener(myJmsConnectionListener);
            // 创建会话。
            // Session.CLIENT_ACKNOWLEDGE: 收到消息后，需要手动调用message.acknowledge()。
            // Session.AUTO_ACKNOWLEDGE: SDK自动ACK（推荐）。
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            connection.start();
            // 创建Receiver连接。
            MessageConsumer consumer = session.createConsumer(queue);
            consumer.setMessageListener(listener);
        }
    }

    @NotNull
    private static Context getContext(AMQPConfig amqp) throws NamingException {
        String connectionUrl = "failover:(amqps://" + amqp.getHost() + ":5671?amqp.idleTimeout=80000)"
                + "?failover.reconnectDelay=30";
        Hashtable<String, String> hashtable = new Hashtable<>();
        hashtable.put("connectionfactory.SBCF", connectionUrl);
        hashtable.put("queue.QUEUE", "default");
        hashtable.put(Context.INITIAL_CONTEXT_FACTORY, "org.apache.qpid.jms.jndi.JmsInitialContextFactory");
        return new InitialContext(hashtable);
    }

    /**
     * @Description: 连接监听类
     * @Creator: 93000
     * @Date: 2022/4/11 15:58
     * @Param:
     * @return:
     */
    private static final JmsConnectionListener myJmsConnectionListener = new JmsConnectionListener() {
        /**
         * 连接成功建立。
         */
        @Override
        public void onConnectionEstablished(URI remoteURI) {
            logger.info("连接建立成功(onConnectionEstablished), remoteUri:{}", remoteURI);
        }

        /**
         * 尝试过最大重试次数之后，最终连接失败。
         */
        @Override
        public void onConnectionFailure(Throwable error) {
            logger.error("连接失败(onConnectionFailure), {}", error.getMessage());
        }

        /**
         * 连接中断。
         */
        @Override
        public void onConnectionInterrupted(URI remoteURI) {
            logger.info("连接中断(onConnectionInterrupted), remoteUri:{}", remoteURI);
        }

        /**
         * 连接中断后又自动重连上。
         */
        @Override
        public void onConnectionRestored(URI remoteURI) {
            logger.info("连接中断重连成功(onConnectionRestored), remoteUri:{}", remoteURI);
        }

        @Override
        public void onInboundMessage(JmsInboundMessageDispatch envelope) {}

        @Override
        public void onSessionClosed(Session session, Throwable cause) {}

        @Override
        public void onConsumerClosed(MessageConsumer consumer, Throwable cause) {}

        @Override
        public void onProducerClosed(MessageProducer producer, Throwable cause) {}
    };

    /**
     * 计算签名，password组装方法，请参见AMQP客户端接入说明文档。
     */
    private static String doSign(String toSignString, String secret, String signMethod) throws Exception {
        SecretKeySpec signingKey = new SecretKeySpec(secret.getBytes(), signMethod);
        Mac mac = Mac.getInstance(signMethod);
        mac.init(signingKey);
        byte[] rawHmac = mac.doFinal(toSignString.getBytes());
        return Base64.encodeBase64String(rawHmac);
    }
}